package com.atguigu.userprofile.task;

import com.atguigu.userprofile.beans.TagInfo;
import com.atguigu.userprofile.constant.ConstCode;
import com.atguigu.userprofile.dao.TagInfoDao;
import com.atguigu.userprofile.utils.ClickhouseUtil;
import com.atguigu.userprofile.utils.MyPropsUtil;
import org.apache.spark.SparkConf;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SaveMode;
import org.apache.spark.sql.SparkSession;

import java.util.List;
import java.util.Properties;
import java.util.stream.Collectors;

/**
 * 将Hive中的标签宽表迁移到ClickHouse中
 * 1、获取传入的参数
 * 2、明确有哪些标签宽表
 * 3、创建标签宽表
 * 4、从Hive中查出数据
 * 5、导出到ClickHouse中
 */
public class TaskExport {
    public static void main(String[] args) {
        // 1、获取传入的参数
        String taskId = args[0];
        String businessDate = args[1];

        // 2、明确有哪些标签宽表
        List<TagInfo> tagInfos = TagInfoDao.selectTagInfosWithEnable();

        // 3、创建标签表
        /*
            create table [ckDbName].[tableName]
            (
                uid String,
                [tagColumns]
            )
            engine=MergeTree
            order by (uid)

         */
        String ckDbName = MyPropsUtil.get(ConstCode.USER_PROFILE_DBNAME);
        String tableName = "up_merge_" + businessDate.replace("-", "");
        // 根据要计算的标签生成宽表的列
        String tagColumns = tagInfos.stream().map(
                tagInfo -> tagInfo.getTagCode().toLowerCase() + " String"
        ).collect(Collectors.joining(" , "));

        String createTable = " create table " + ckDbName + "." + tableName +
                " (" +
                " uid String , " + tagColumns +
                " )" +
                " engine = MergeTree " +
                " order by (uid)";

        // System.out.println(createTable);

        String dropTable = "drop table if exists " + ckDbName + "." + tableName;
        ClickhouseUtil.executeSQL(dropTable);
        ClickhouseUtil.executeSQL(createTable);

        // 4、从Hive中查出数据
        String upDbName = MyPropsUtil.get(ConstCode.USER_PROFILE_DBNAME);
        String selectColumns = tagInfos.stream().map(
                tagInfo -> tagInfo.getTagCode().toLowerCase()
        ).collect(Collectors.joining(" , "));
        String selectFromHive = "select uid , " + selectColumns + " from " + upDbName + "." + tableName;
        SparkConf conf = new SparkConf().setAppName("task_export_app");//.setMaster("local[*]");
        SparkSession spark = SparkSession.builder().config(conf).enableHiveSupport().getOrCreate();
        Dataset<Row> dataset = spark.sql(selectFromHive);

        // 5、导出到ClickHouse中
        String clickhouseUrl = MyPropsUtil.get(ConstCode.CLICKHOUSE_URL);
        dataset.write().mode(SaveMode.Append)
                .option("driver", "ru.yandex.clickhouse.ClickHouseDriver")
                .option("batchsize", 200)
                .option("isolationLevel", "NONE")
                .option("numPartitions", 4)
                .jdbc(clickhouseUrl, tableName, new Properties());

    }
}
